/*
 * Copyright 2007 Stefan Arentz <http://stefan.arentz.nl>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.sateh.spaces.simple.store;

import com.sateh.spaces.simple.SimpleSpaceObject;
import com.sateh.spaces.simple.SimpleSpaceTransactionId;
import com.sateh.spaces.simple.matcher.ObjectMatcher;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.util.*;

public class SimpleSpaceObjectStore implements ObjectStore
{
    private final static Log log = LogFactory.getLog(SimpleSpaceObjectStore.class);

    /**
     *
     */

    private final Set<SimpleSpaceObject> objects = new HashSet<SimpleSpaceObject>();

    /**
     *
     */
    
    private final Map<SimpleSpaceTransactionId,Set<SimpleSpaceObject>> objectsPerTransaction = new HashMap<SimpleSpaceTransactionId, Set<SimpleSpaceObject>>();

    /**
     *
     */

    public void storeObject(SimpleSpaceObject object, SimpleSpaceTransactionId transactionId)
    {
        synchronized (this)
        {
            if (transactionId != null) {
                Set<SimpleSpaceObject> objectsInTransaction = getObjectsInTransaction(transactionId);
                if (objectsInTransaction == null) {
                    objectsInTransaction = new HashSet<SimpleSpaceObject>();
                    objectsPerTransaction.put(transactionId, objectsInTransaction);
                }
                objectsInTransaction.add(object);
            } else {
                objects.add(object);
            }
        }
    }
    
    /**
     * @see com.sateh.spaces.simple.store.ObjectStore#matchAndTakeObject(com.sateh.spaces.simple.matcher.ObjectMatcher)
     */

    public SimpleSpaceObject matchAndTakeObject(ObjectMatcher matcher)
    {
        synchronized(objects) {
            for (SimpleSpaceObject simpleSpaceObject : objects) {
                if (!simpleSpaceObject.isExpired()) {
                    if (matcher.willMatch(simpleSpaceObject)) {
                        objects.remove(simpleSpaceObject);
                        return simpleSpaceObject;
                    }
                }
            }
        }

        return null;
    }

    /**
     * @see com.sateh.spaces.simple.store.ObjectStore#matchAndReadObject(ObjectMatcher)
     */

    public SimpleSpaceObject matchAndReadObject(ObjectMatcher matcher)
    {
        SimpleSpaceTransactionId transactionId = matcher.getTransactionId();
        if (transactionId != null) {
            Set<SimpleSpaceObject> objectsInTransaction = getObjectsInTransaction(transactionId);
            if (objectsInTransaction != null) {
                for (SimpleSpaceObject simpleSpaceObject : objectsInTransaction) {
                    if (!simpleSpaceObject.isExpired()) {
                        if (matcher.willMatch(simpleSpaceObject)) {
                            return simpleSpaceObject;
                        }
                    }
                }
            }
        }

        synchronized(objects) {
            for (SimpleSpaceObject simpleSpaceObject : objects) {
                if (!simpleSpaceObject.isExpired()) {
                    if (matcher.willMatch(simpleSpaceObject)) {
                        if (transactionId != null) {
                            // If this object was read from the global space within a transaction, then it becomes
                            // part of the transactional space.
                            objects.remove(simpleSpaceObject);
                            Set<SimpleSpaceObject> objectsInTransaction = getObjectsInTransaction(transactionId);
                            if (objectsInTransaction == null) {
                                objectsInTransaction = new HashSet<SimpleSpaceObject>();
                                objectsPerTransaction.put(transactionId, objectsInTransaction);
                            }
                            objectsInTransaction.add(simpleSpaceObject);
                        }
                        return simpleSpaceObject;
                    }
                }
            }
        }

        return null;
    }

    /**
     * @see com.sateh.spaces.simple.store.ObjectStore#cancelObjectLease(java.util.UUID)
     */

    public boolean cancelObjectLease(UUID objectId)
    {
        boolean cancelled = false;

        synchronized (objects)
        {
            for (SimpleSpaceObject object : objects) {
                if (!object.isExpired()) {
                    if (object.getId().compareTo(objectId) == 0) {
                        objects.remove(object);
                        cancelled = true;
                        break;
                    }
                }
            }
        }

        return cancelled;        
    }

    /**
     * @see com.sateh.spaces.simple.store.ObjectStore#renewObjectLease(java.util.UUID, long)
     */

    public boolean renewObjectLease(UUID objectId, long timeToLive)
    {
        synchronized (objects)
        {
            for (SimpleSpaceObject object : objects) {
                if (!object.isExpired()) {
                    if (object.getId().compareTo(objectId) == 0) {
                        object.renew(timeToLive);
                        return true;
                    }
                }
            }
        }

        return false;        
    }

    /**
     * @see com.sateh.spaces.simple.store.ObjectStore#matchAndCount(com.sateh.spaces.simple.matcher.ObjectMatcher, com.sateh.spaces.simple.SimpleSpaceTransactionId)  
     */

    public long matchAndCount(ObjectMatcher matcher, SimpleSpaceTransactionId transactionId)
    {
        long count = 0;

        synchronized(this)
        {
            for (SimpleSpaceObject simpleSpaceObject : objects) {
                if (!simpleSpaceObject.isExpired()) {
                    if (matcher.willMatch(simpleSpaceObject)) {
                        count++;
                    }
                }
            }

            if (transactionId != null) {
                Set<SimpleSpaceObject> objectsInTransaction = getObjectsInTransaction(transactionId);
                if (objectsInTransaction != null) {
                    for (SimpleSpaceObject simpleSpaceObject : objectsInTransaction) {
                        if (!simpleSpaceObject.isExpired()) {
                            if (matcher.willMatch(simpleSpaceObject)) {
                                count++;
                            }
                        }
                    }                    
                }
            }
        }

        return count;
    }

    private Set<SimpleSpaceObject> getObjectsInTransaction(SimpleSpaceTransactionId transactionId)
    {
        return objectsPerTransaction.get(transactionId);
    }

    public void mergeObjectsInTransaction(SimpleSpaceTransactionId transactionId)
    {
        if (log.isDebugEnabled()) {
            log.debug("mergeObjectsInTransaction; transactionId=" + transactionId);
        }

        synchronized(this)
        {
            Set<SimpleSpaceObject> objectsInTransaction = getObjectsInTransaction(transactionId);
            if (objectsInTransaction != null) {
                objects.addAll(objectsInTransaction);
            }
        }
    }

    public void releaseObjectsInTransaction(SimpleSpaceTransactionId transactionId)
    {
        if (log.isDebugEnabled()) {
            log.debug("releaseObjectsInTransaction; transactionId=" + transactionId);
        }
    }
}
